﻿using DelayedQueue.Data;
using DelayedQueue.Kafka;
using DelayedQueue.Models;

namespace DelayedQueue
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("=== Kafka + MySQL + 时间轮 延时队列系统 ===");
            Console.WriteLine("1. 启动生产者 (发送延时任务)");
            Console.WriteLine("2. 启动消费者 (处理延时任务)");
            Console.Write("请选择操作: ");

            var key = Console.ReadKey();
            Console.WriteLine();

            try
            {
                // 读取配置
                var kafkaBootstrapServers = "192.168.137.10:9092";
                var kafkaTopic = "delayed-queue";
                var kafkaConsumerGroupId = "delayed-queue-group";
                var mySqlConnectionString = "server=192.168.137.10,3306;database=delayed_queue;uid=root;pwd=admin110;SslMode=None";
                var slotCount = 60;//槽位数量
                var tickDurationMs = 1000;//每个槽位的毫秒数

                switch (key.KeyChar)
                {
                    case '1':
                        RunProducer(kafkaBootstrapServers, kafkaTopic);
                        break;
                    case '2':
                        RunConsumer(
                            kafkaBootstrapServers,
                            kafkaTopic,
                            kafkaConsumerGroupId,
                            mySqlConnectionString,
                            slotCount,
                            tickDurationMs);
                        break;
                    default:
                        Console.WriteLine("无效选择");
                        break;
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine($"程序错误: {ex.Message}");
                Console.WriteLine(ex.StackTrace);
            }
        }

        /// <summary>
        /// 运行生产者
        /// </summary>
        static void RunProducer(string bootstrapServers, string topic)
        {
            using (var producer = new KafkaProducer(bootstrapServers, topic))
            {
                Console.WriteLine("生产者模式 - 按Ctrl+C退出");
                try
                {
                    while (true)
                    {
                        Console.Write("请输入任务内容: ");
                        var content = Console.ReadLine();

                        Console.Write("请输入延迟时间(秒): ");
                        if (int.TryParse(Console.ReadLine(), out int delaySeconds) && delaySeconds > 0)
                        {
                            var task = new DelayedTask
                            {
                                Content = content,
                                ExecuteTimeUtc = DateTime.UtcNow.AddSeconds(delaySeconds)
                            };

                            var result = producer.SendDelayedTaskAsync(task).Result;
                            Console.WriteLine($"任务已发送，分区: {result.Partition}, 偏移量: {result.Offset}");
                        }
                        else
                        {
                            Console.WriteLine("请输入有效的延迟时间");
                        }
                    }
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"发送任务出错: {ex.Message}");
                }
            }
        }

        /// <summary>
        /// 运行消费者
        /// </summary>
        static void RunConsumer(
            string bootstrapServers,
            string topic,
            string groupId,
            string connectionString,
            int slotCount,
            int tickDurationMs)
        {
            // 创建持久化管理器
            using (var persistence = new MySQLPersistence(connectionString))
            {
                // 创建任务执行器
                Action<DelayedTask> taskExecutor = task =>
                {
                    try
                    {
                        Console.WriteLine($"[{DateTime.Now:yyyy-mm-dd HH:mm:ss.fff}] 执行任务 {task.TaskId}: {task.Content}");

                        // 这里可以添加实际的任务处理逻辑

                        // 更新任务状态为已完成
                        persistence.UpdateTaskStatus(task.TaskId, Models.TaskStatus.Completed);
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine($"执行任务 {task.TaskId} 失败: {ex.Message}");
                        persistence.UpdateTaskStatus(task.TaskId, Models.TaskStatus.Failed);
                    }
                };

                // 创建时间轮
                using (var timingWheel = new TimingWheel.TimingWheel(slotCount, tickDurationMs, taskExecutor))
                {
                    // 创建Kafka消费者
                    using (var consumer = new KafkaConsumer(
                        bootstrapServers,
                        topic,
                        groupId,
                        timingWheel,
                        persistence))
                    {
                        Console.WriteLine("消费者模式 - 按Ctrl+C退出");

                        // 启动时间轮
                        timingWheel.Start();

                        // 加载未完成的任务
                        var pendingTasks = persistence.GetPendingTasks();
                        Console.WriteLine($"加载 {pendingTasks.Count} 个未完成的任务");
                        foreach (var task in pendingTasks)
                        {
                            timingWheel.AddTask(task);
                        }

                        // 启动消费者
                        consumer.StartConsuming();

                        // 等待退出信号
                        var exitEvent = new ManualResetEvent(false);
                        Console.CancelKeyPress += (sender, e) =>
                        {
                            e.Cancel = true;
                            exitEvent.Set();
                        };

                        exitEvent.WaitOne();

                        // 清理资源
                        consumer.StopConsuming();
                        timingWheel.Stop();
                    }
                }
            }
        }
    }
}
